

package com.custom.serde;


import com.custom.source.ClinkEnumCheckpoint;
import com.custom.source.ClinkSourceSplit;
import lombok.val;
import org.apache.flink.core.io.SimpleVersionedSerializer;

import java.io.IOException;

/**
 * 枚举分片保存点序列化器
 *
 * @param <SplitT>   分片类型
 * @param <EnumChkT> 枚举分片类型
 */
public class DefaultEnumCheckpointSerializer<SplitT extends ClinkSourceSplit, EnumChkT extends ClinkEnumCheckpoint<SplitT>> implements SimpleVersionedSerializer<EnumChkT> {

    @Override
    public int getVersion() {
        return 0;
    }

    @Override
    public byte[] serialize(EnumChkT clinkSplitEnumState) throws IOException {
        return SerializerUtils.serialize(clinkSplitEnumState);
    }

    @Override
    public EnumChkT deserialize(int version, byte[] serialized) throws IOException {
        if (version != getVersion()) {
            throw new IOException("Unrecognized version: " + version);
        }
        @SuppressWarnings("unchecked")
        val enumChk = (EnumChkT) SerializerUtils.deserialize(serialized);
        return enumChk;
    }
}
